package com.atguigu.gmall.realtime.app.func;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.gmall.realtime.beans.TableProcess;
import com.atguigu.gmall.realtime.common.GmallConfig;
import org.apache.flink.api.common.state.BroadcastState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.state.ReadOnlyBroadcastState;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.co.BroadcastProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.util.*;

/**
 * Author: Felix
 * Date: 2022/1/17
 * Desc: 业务数据动态分流实现
 */
public class TableProcessFunction extends BroadcastProcessFunction<JSONObject, String, JSONObject> {
    private OutputTag<JSONObject> dimTag;
    private MapStateDescriptor<String, TableProcess> mapStateDescriptor;
    private Connection conn;

    public TableProcessFunction(OutputTag<JSONObject> dimTag, MapStateDescriptor<String, TableProcess> mapStateDescriptor) {
        this.dimTag = dimTag;
        this.mapStateDescriptor = mapStateDescriptor;
    }

    @Override
    public void open(Configuration parameters) throws Exception {
        //注册驱动
        Class.forName("org.apache.phoenix.jdbc.PhoenixDriver");
        //创建连接
        conn = DriverManager.getConnection(GmallConfig.PHOENIX_SERVER);
    }

    //处理主流中的业务数据
    @Override
    public void processElement(JSONObject jsonObj, ReadOnlyContext ctx, Collector<JSONObject> out) throws Exception {
        //获取广播状态
        ReadOnlyBroadcastState<String, TableProcess> broadcastState = ctx.getBroadcastState(mapStateDescriptor);

        //获取表名以及操作类型
        String tableName = jsonObj.getString("table");
        String type = jsonObj.getString("type");

        //注意：如果使用maxwell的bootstrap读取历史数据的话，那么操作类型是bootstrap-insert，我们统一设置为insert
        if (type.equals("bootstrap-insert")) {
            type = "insert";
            jsonObj.put("type", type);
        }

        //拼接从状态中获取配置信息的key
        String key = tableName + ":" + type;

        //根据key到状态中获取对应的配置信息
        TableProcess tableProcess = broadcastState.get(key);

        if(tableProcess != null){
            //说明在配置表中找到了当前处理的数据对应的配置信息  根据配置信息中的sinkType属性，进行分流
            String sinkType = tableProcess.getSinkType();
            //不管是主流事实数据，还是侧输出流维度数据，在向下游传递之前，都需要携带上目的地
            String sinkTable = tableProcess.getSinkTable();
            jsonObj.put("sink_table",sinkTable);

            //过滤字段
            JSONObject dataJsonObj = jsonObj.getJSONObject("data");
            String sinkColumns = tableProcess.getSinkColumns();
            filterColumn(dataJsonObj,sinkColumns);

            if(TableProcess.SINK_TYPE_HBASE.equals(sinkType)){
                //维度数据  ---放到维度侧输出流
                ctx.output(dimTag,jsonObj);
            }else if(TableProcess.SINK_TYPE_KAFKA.equals(sinkType)){
                //事实数据 ---放到主流中
                out.collect(jsonObj);
            }
        }else{
            //说明在配置表中没有当前处理的数据对应的配置信息
            System.out.println("No this key in TableProcess:" + key);
        }
    }

    //处理广播流中的配置数据
    //jsonStr  :{"database":"","table":"","type":"","data":{配置表中的一条配置信息}}
    @Override
    public void processBroadcastElement(String jsonStr, Context ctx, Collector<JSONObject> out) throws Exception {
        //为了处理方便，将jsonStr转换为jsonObj
        JSONObject jsonObj = JSON.parseObject(jsonStr);
        //从json对象中获取data属性--->配置表中的一条配置信息
        JSONObject dataJsonObj = jsonObj.getJSONObject("data");
        //将读取到的一条配置信息封装为TableProcess对象
        TableProcess tableProcess = dataJsonObj.toJavaObject(TableProcess.class);
        //根据配置对象获取相关的属性
        //获取业务表名
        String sourceTable = tableProcess.getSourceTable();
        //获取操作类型
        String operateType = tableProcess.getOperateType();
        //获取输出类型    hbase|kafka
        String sinkType = tableProcess.getSinkType();
        //获取目的地表名或者主题名
        String sinkTable = tableProcess.getSinkTable();
        //获取保留字段（建表字段）
        String sinkColumns = tableProcess.getSinkColumns();
        //获取建表主键
        String sinkPk = tableProcess.getSinkPk();
        //获取建表扩展语句
        String sinkExtend = tableProcess.getSinkExtend();

        //将表名以及操作类型拼接为广播状态中存储信息的key
        String key = sourceTable + ":" + operateType;

        //判断当前从配置表中读取到的配置信息是事实还是维度，如果是维度配置，提前将维度表创建出来
        if(TableProcess.SINK_TYPE_HBASE.equals(sinkType)&&"insert".equals(operateType)){
            checkTable(sinkTable,sinkColumns,sinkPk,sinkExtend);
        }

        //将读取到的一条配置信息放到广播状态中
        BroadcastState<String, TableProcess> broadcastState = ctx.getBroadcastState(mapStateDescriptor);
        broadcastState.put(key,tableProcess);
    }

    //创建维度表
    private void checkTable(String tableName, String fields, String pk, String ext) {
        //空值处理
        if(pk == null){
            pk = "id";
        }
        if(ext == null){
            ext = "";
        }

        //拼接建表语句
        StringBuilder createSql = new StringBuilder("create table if not exists "+ GmallConfig.HBASE_SCHEMA+"."+tableName+"(");
        String[] fieldArr = fields.split(",");
        for (int i = 0; i < fieldArr.length; i++) {
            String field = fieldArr[i];
            //判断是否为主键字段
            if(pk.equals(field)){
                createSql.append(field + " varchar primary key ");
            }else{
                createSql.append(field + " varchar ");
            }

            if(i < fieldArr.length - 1){
                createSql.append(",");
            }
        }
        createSql.append(")" + ext);

        System.out.println("在phoenix中执行的建表语句：" + createSql);

        //获取数据库操作对象
        PreparedStatement ps = null;
        try {
            ps = conn.prepareStatement(createSql.toString());
            //执行SQL语句
            ps.execute();
        } catch (Exception e) {
            e.printStackTrace();
            throw  new RuntimeException("在phoenix中创建维度表发生了异常~~~");
        }finally {
            //释放资源
            if(ps != null){
                try {
                    ps.close();
                } catch (SQLException e) {
                    e.printStackTrace();
                }
            }
        }
    }

    //过滤字段
    // dataJsonObj :{"tm_name":"aabb","logo_url":"aaa","id":12}
    //sinkColumns : id,tm_name
    private void filterColumn(JSONObject dataJsonObj, String sinkColumns) {
        //将保留字段字符串进行分割
        String[] fieldArr = sinkColumns.split(",");
        List<String> fieldList = Arrays.asList(fieldArr);

        Set<Map.Entry<String, Object>> entrySet = dataJsonObj.entrySet();
        entrySet.removeIf(entry->!fieldList.contains(entry.getKey()));
    }
}
